package com.homework.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;

public class UserTable {

    Configuration conf = null;
    Connection conn = null;
    Scanner sc = new Scanner(System.in);

    @Before
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "centos7-1,centos7-2");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        conn = ConnectionFactory.createConnection(conf);
    }

    @Test
    public void createTable() throws IOException {
        HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();

        HTableDescriptor user_rel = new HTableDescriptor(TableName.valueOf("user_rel"));

        user_rel.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(user_rel);

        System.out.println("user_rel表创建成功");
    }

    @Test
    public void insertData() throws IOException {
        //获取表对象
        Table user_rel = conn.getTable(TableName.valueOf("user_rel"));

        List<Put> puts = new ArrayList<Put>();//指定一个集合，批量插入数据

        //使用for循环，每一个用户插入10个好友id
        Put put1 = new Put(Bytes.toBytes("uid1"));//指定rowkey
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid4"),Bytes.toBytes("uid4"));
        puts.add(put1);


        //uid2的用户
        Put put2 = new Put(Bytes.toBytes("uid2"));//指定rowkey
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid4"),Bytes.toBytes("uid4"));
        puts.add(put2);


        //uid3的用户
        Put put3 = new Put(Bytes.toBytes("uid3"));//指定rowkey
        put3.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put3.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put3.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid4"),Bytes.toBytes("uid4"));
        puts.add(put3);


        user_rel.put(puts);
        user_rel.close();

        System.out.println("数据插入成功");
    }

        public void deleteFriend() throws IOException {
        //获取表对象
        Table user_rel = conn.getTable(TableName.valueOf("user_rel"));

        System.out.println("输入登录的账号");
        String rowkey = sc.nextLine();
            System.out.println(rowkey);
        //展示所有好友
        System.out.println("该用户下的好友如下：");
        Get get = new Get(Bytes.toBytes(rowkey));
        Result result = user_rel.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            final String rk = Bytes.toString(CellUtil.cloneRow(cell));
            final String f = Bytes.toString(CellUtil.cloneFamily(cell));
            final String col = Bytes.toString(CellUtil.cloneQualifier(cell));
            final String value = Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println("rowkey-->" + rk + "--;cf-->" + f + "---;column--->" + col + "--;value-->" + value);
        }


        System.out.println("删除哪个好友");
        String column = sc.nextLine();

        //删除登录账号的好友
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        delete.deleteColumn(Bytes.toBytes("friends"),Bytes.toBytes(column));
        user_rel.delete(delete);


        //通过删除的好友uid作为rowkey进行查询，再删除登录账号的uid
        Delete delete2 = new Delete(Bytes.toBytes(column));
        delete2.deleteColumn(Bytes.toBytes("friends"),Bytes.toBytes(rowkey));
        user_rel.delete(delete2);

        //展示删除好友后的所有好友
//        System.out.println("该用户下的好友如下：");
//        Get get1 = new Get(Bytes.toBytes(rowkey));
//        Result result1 = user_rel.get(get1);
//        Cell[] cells1 = result.rawCells();
//        for (Cell cell : cells1) {
//            final String rk = Bytes.toString(CellUtil.cloneRow(cell));
//            final String f = Bytes.toString(CellUtil.cloneFamily(cell));
//            final String col = Bytes.toString(CellUtil.cloneQualifier(cell));
//            final String value = Bytes.toString(CellUtil.cloneValue(cell));
//
//            System.out.println("rowkey-->" + rk + "--;cf-->" + f + "---;column--->" + col + "--;value-->" + value);
//        }
        user_rel.close();
        System.out.println("双向好友删除成功");
    }

    //释放连接
    @After
    public void release() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }



}
